package com.lingjtx.iot.network.config;

import com.lingjtx.common.core.constant.MqConst;
import com.lingjtx.common.core.handle.EventMessageHandle;
import com.lingjtx.iot.network.common.ErrorMessageHandler;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import java.util.Objects;

/**
 * MQTT 接收端配置
 */
@Slf4j
@Order(3)
@Configuration
public class MqttSubscriberConfig {

    private final MqttConfig mqttConfig;
    private final EventMessageHandle messageHandle;

    public MqttSubscriberConfig(MqttConfig mqttConfig,
                                @Qualifier("mqttReportData") EventMessageHandle messageHandle) {
        this.mqttConfig = mqttConfig;
        this.messageHandle = messageHandle;
    }

    @SneakyThrows
    @Bean
    public MqttPahoClientFactory subscriberClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(mqttConfig.mqttConnectOptions());
        return factory;
    }

    @Bean("mqttSubscriberAdapter")
    public MqttPahoMessageDrivenChannelAdapter mqttSubscriberAdapter() {
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                "subscriber_" + mqttConfig.mqttClientId(),
                subscriberClientFactory(),
                MqConst.MqttSubscriberTopic.SHARE_GROUP + MqConst.MqttSubscriberTopic.MQTT_TOPIC_LINGJTX,
                MqConst.MqttSubscriberTopic.SHARE_GROUP + MqConst.MqttSubscriberTopic.MQTT_TOPIC_SYS_SERVER,
                MqConst.MqttSubscriberTopic.SHARE_GROUP + MqConst.MqttSubscriberTopic.MQTT_TOPIC_COMMUNICATION_MANAGEMENT
        );
        adapter.setQos(mqttConfig.mqttProperties().getQos());
        adapter.setOutputChannel(mqttSubscriberChannel());
        adapter.setErrorChannel(mqttSubscriberErrorChannel());
        adapter.setTaskScheduler(mqttConfig.taskScheduler());
        adapter.setDisconnectCompletionTimeout(mqttConfig.mqttProperties().getTimeout());
        return adapter;
    }


    @Bean
    public MessageChannel mqttSubscriberChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttSubscriberChannel")
    public MessageHandler subscriberMessageHandler() {
        return message -> {
            String payload = message.getPayload().toString();
            String topic = Objects.requireNonNull(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC)).toString();
            log.debug("====>> 主题: {}, 消息内容: {}", topic, payload);
            messageHandle.process(topic, payload);
        };
    }

    @Bean
    public MessageChannel mqttSubscriberErrorChannel() {
        return new DirectChannel();
    }

    @Bean
    @ServiceActivator(inputChannel = "mqttSubscriberErrorChannel")
    public MessageHandler subscriberErrorHandler() {
        return new ErrorMessageHandler(mqttConfig.retryTemplate());
    }
}
